/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.streaming;

import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.ipig.computing.commons.streaming.StreamExecutorService;
import org.ipig.computing.spark.conf.SparkExecutorConf;
import org.ipig.constants.ResultStatus;

/**
 * SparkStreamService
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
public interface SparkStreamService<T> extends StreamExecutorService<SparkExecutorConf, SparkSession> {
    public default ResultStatus execute() {
        ResultStatus result = ResultStatus.SUCCESS;
        SparkSession sparkSession = getSession();
        SparkContext sparkContext = sparkSession.sparkContext();
        JavaSparkContext sc = new JavaSparkContext(sparkContext);
        JavaStreamingContext ssc = getJavaStreamingContext(sc);
        try {
            //1全局初始化
            init(ssc);
            //2获取数据
            JavaInputDStream<T> stream = fromInputDStream(ssc);
            doStream(stream);
            // 将关闭处理程序附加到catch control-c
            Runtime.getRuntime().addShutdownHook(new Thread("Spark Stream-shutdown-hook") {
                @Override
                public void run() {
                    if (sparkSession != null) {
                        sparkSession.close();
                    }
                    SparkSession.clearActiveSession();
                    if (ssc != null) {
                        ssc.close();
                    }
                }
            });

            try {
                ssc.start();
                ssc.awaitTermination();
            } catch (Throwable e) {
                System.exit(1);
            }
            System.exit(0);
        } catch (Exception e) {
            e.printStackTrace();
            result = ResultStatus.FAILURE;
        }
        return result;
    }

    /**
     * 处理流
     *
     * @param stream
     */
    public void doStream(JavaInputDStream<T> stream);


    /**
     * 获取 StreamingContext
     *
     * @param sc JavaSparkContext
     * @return JavaStreamingContext
     */
    public JavaStreamingContext getJavaStreamingContext(JavaSparkContext sc);

    /**
     * 全局初始化
     * <li>更改SparkContext,如设置日志，检查点，注册函数、累加器、广播变量等</li>
     *
     * @param jsc JavaStreamingContext
     */
    public void init(JavaStreamingContext jsc);

    /**
     * 获取输入数据流InputDStream
     *
     * @param jsc JavaStreamingContext
     * @return JavaInputDStream<T>
     */
    public JavaInputDStream<T> fromInputDStream(JavaStreamingContext jsc);

}

